//! # PostgreSQL数据库存储演示
//!
//! 演示基于PostgreSQL的消息持久化功能
//! 
//! 注意: 此演示需要PostgreSQL数据库，如果没有可用数据库，将显示配置和使用方法

use rustcloud_bus::{
    db_storage::{DatabaseStorageConfig, DatabaseType, PostgreSQLPool, DatabaseStorage},
    persistence::{MessageStorage, QueryCriteria, MessageStatus, OrderBy, SortOrder},
    message::Message,
    error::MessageBusResult,
};
use serde_json::json;
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt, EnvFilter};

async fn demonstrate_postgresql_storage() -> MessageBusResult<()> {
    println!("🐘 PostgreSQL数据库存储演示\n");

    // 1. 创建数据库配置
    println!("=== 1. 数据库配置 ===");
    let config = DatabaseStorageConfig {
        database_type: DatabaseType::PostgreSQL,
        connection_url: std::env::var("DATABASE_URL")
            .unwrap_or_else(|_| "postgresql://username:password@localhost:5432/rustcloud_bus".to_string()),
        max_connections: 10,
        min_connections: 2,
        connect_timeout_secs: 30,
        idle_timeout_secs: 600,
        max_query_timeout_secs: 30,
        table_prefix: "demo_".to_string(),
        auto_migrate: true,
        enable_pool_monitoring: true,
        batch_size_limit: 1000,
    };

    println!("✅ 数据库配置:");
    println!("   • 数据库类型: {:?}", config.database_type);
    println!("   • 连接URL: {}", mask_database_url(&config.connection_url));
    println!("   • 最大连接数: {}", config.max_connections);
    println!("   • 表前缀: {}", config.table_prefix);

    // 检查是否有可用的数据库连接
    if std::env::var("DATABASE_URL").is_err() && !is_local_postgres_available().await {
        println!("\n⚠️  没有检测到可用的PostgreSQL数据库");
        println!("要运行此演示，请:");
        println!("   1. 安装PostgreSQL数据库");
        println!("   2. 创建数据库: CREATE DATABASE rustcloud_bus;");
        println!("   3. 设置环境变量: DATABASE_URL=postgresql://user:pass@localhost:5432/rustcloud_bus");
        println!("   4. 重新运行演示程序");
        
        demonstrate_sql_schema(&config);
        return Ok(());
    }

    // 2. 创建连接池
    println!("\n=== 2. 创建PostgreSQL连接池 ===");
    let pool = match PostgreSQLPool::new(config.clone()).await {
        Ok(pool) => {
            println!("✅ PostgreSQL连接池创建成功");
            pool
        }
        Err(e) => {
            println!("❌ PostgreSQL连接池创建失败: {}", e);
            println!("这通常是因为:");
            println!("   • PostgreSQL服务未运行");
            println!("   • 连接URL不正确");
            println!("   • 数据库不存在");
            println!("   • 用户名密码错误");
            
            demonstrate_sql_schema(&config);
            return Err(e);
        }
    };

    // 3. 执行数据库迁移
    println!("\n=== 3. 执行数据库迁移 ===");
    pool.migrate().await?;
    println!("✅ 数据库表和索引创建完成");

    // 4. 创建存储实例
    println!("\n=== 4. 创建数据库存储 ===");
    let storage = DatabaseStorage::new(pool, config.clone());
    
    // 健康检查
    storage.health_check().await?;
    println!("✅ 数据库存储健康检查通过");

    // 5. 存储消息演示
    println!("\n=== 5. 消息存储演示 ===");
    let test_messages = vec![
        Message::new("user.events.registered", json!({
            "user_id": "user_001",
            "email": "alice@example.com",
            "username": "alice"
        })),
        Message::new("order.events.created", json!({
            "order_id": "order_001", 
            "user_id": "user_001",
            "amount": 99.99,
            "currency": "USD"
        })),
        Message::new("payment.events.processed", json!({
            "payment_id": "pay_001",
            "order_id": "order_001", 
            "status": "completed",
            "amount": 99.99
        })),
    ];

    let mut stored_ids = Vec::new();
    for (i, message) in test_messages.into_iter().enumerate() {
        let id = storage.store(message.clone()).await?;
        stored_ids.push(id.clone());
        println!("  📝 消息 #{}: {} -> {}", i + 1, message.topic(), &id[..8]);
    }

    // 6. 查询消息演示
    println!("\n=== 6. 查询消息演示 ===");
    
    // 查询所有消息
    let all_messages = storage.query(QueryCriteria::default()).await?;
    println!("  📊 数据库中总消息数: {}", all_messages.len());
    
    // 按主题查询
    let user_events = storage.query(QueryCriteria {
        topic_pattern: Some("user.events.*".to_string()),
        ..Default::default()
    }).await?;
    println!("  👤 用户事件消息: {}", user_events.len());
    
    // 按状态查询
    let pending_messages = storage.query(QueryCriteria {
        status: Some(MessageStatus::Pending),
        order_by: Some(OrderBy::StoredAt(SortOrder::Descending)),
        limit: Some(5),
        ..Default::default()
    }).await?;
    println!("  ⏳ 待处理消息 (最新5条):");
    for (i, record) in pending_messages.iter().enumerate() {
        println!("    {}. {} - {} - {}", 
            i + 1, 
            record.message.topic(),
            &record.message.id()[..8],
            record.stored_at.format("%H:%M:%S")
        );
    }

    // 7. 消息状态更新
    println!("\n=== 7. 消息状态更新演示 ===");
    if !stored_ids.is_empty() {
        let first_id = &stored_ids[0];
        
        // 更新为处理中
        storage.update_status(first_id, MessageStatus::Processing).await?;
        println!("  🔄 消息状态更新: {} -> Processing", &first_id[..8]);
        
        // 更新为已完成
        storage.update_status(first_id, MessageStatus::Completed).await?;
        println!("  ✅ 消息状态更新: {} -> Completed", &first_id[..8]);
    }

    // 8. 批量操作演示
    println!("\n=== 8. 批量操作演示 ===");
    let batch_messages = vec![
        Message::new("batch.test.1", json!({"batch_id": 1, "data": "test1"})),
        Message::new("batch.test.2", json!({"batch_id": 2, "data": "test2"})),
        Message::new("batch.test.3", json!({"batch_id": 3, "data": "test3"})),
    ];
    
    let batch_ids = storage.store_batch(batch_messages).await?;
    println!("  📦 批量存储成功: {} 条消息", batch_ids.len());
    
    // 批量删除测试消息
    storage.delete_batch(batch_ids).await?;
    println!("  🗑️  批量删除成功");

    // 9. 事务演示
    println!("\n=== 9. 事务演示 ===");
    let mut tx = storage.begin_transaction().await?;
    
    let tx_message = Message::new("transaction.test", json!({"tx_id": "tx_001"}));
    let tx_id = tx.store(tx_message).await?;
    println!("  💾 事务中存储消息: {}", &tx_id[..8]);
    
    // 检查事务外不可见
    if storage.get(&tx_id).await?.is_none() {
        println!("  ✅ 事务提交前消息不可见");
    }
    
    tx.commit().await?;
    
    // 检查事务后可见
    if storage.get(&tx_id).await?.is_some() {
        println!("  ✅ 事务提交后消息可见");
    }

    // 10. 统计信息
    println!("\n=== 10. 数据库存储统计 ===");
    let stats = storage.get_stats().await?;
    println!("  📊 存储统计:");
    println!("    • 总消息数: {}", stats.total_messages);
    println!("    • 待处理: {}", stats.pending_messages);
    println!("    • 处理中: {}", stats.processing_messages);
    println!("    • 已完成: {}", stats.completed_messages);
    println!("    • 失败数: {}", stats.failed_messages);
    
    if let Some(oldest) = stats.oldest_message {
        println!("    • 最旧消息: {}", oldest.format("%Y-%m-%d %H:%M:%S"));
    }
    if let Some(newest) = stats.newest_message {
        println!("    • 最新消息: {}", newest.format("%Y-%m-%d %H:%M:%S"));
    }

    // 11. 清理演示
    println!("\n=== 11. 清理过期消息 ===");
    let cleaned = storage.cleanup_expired().await?;
    println!("  🧹 清理过期消息: {} 条", cleaned);

    println!("\n✅ PostgreSQL存储演示完成！");
    
    Ok(())
}

/// 演示SQL架构
fn demonstrate_sql_schema(config: &DatabaseStorageConfig) {
    println!("\n=== SQL架构演示 ===");
    println!("当PostgreSQL可用时，将自动创建以下表结构:\n");
    
    let table_name = format!("{}messages", config.table_prefix);
    println!("📋 消息表: {}", table_name);
    println!(r#"
CREATE TABLE {} (
    id VARCHAR(36) PRIMARY KEY,
    topic VARCHAR(255) NOT NULL,
    payload JSONB NOT NULL,
    headers JSONB,
    status VARCHAR(20) NOT NULL DEFAULT 'pending',
    retry_count INTEGER NOT NULL DEFAULT 0,
    max_retry_count INTEGER NOT NULL DEFAULT 3,
    stored_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW(),
    processed_at TIMESTAMP WITH TIME ZONE,
    expires_at TIMESTAMP WITH TIME ZONE,
    created_by VARCHAR(100),
    trace_id VARCHAR(64),
    correlation_id VARCHAR(64)
);
"#, table_name);

    println!("🔍 索引:");
    println!("  • idx_topic - 主题查询优化");
    println!("  • idx_status - 状态查询优化");
    println!("  • idx_stored_at - 时间排序优化");
    println!("  • idx_expires_at - 过期清理优化");
    println!("  • idx_trace_id - 链路追踪优化");
}

/// 检查本地PostgreSQL是否可用
async fn is_local_postgres_available() -> bool {
    // 简单检查，实际使用中可以更复杂
    std::process::Command::new("pg_isready")
        .arg("-h")
        .arg("localhost")
        .output()
        .map(|output| output.status.success())
        .unwrap_or(false)
}

/// 屏蔽数据库URL中的密码
fn mask_database_url(url: &str) -> String {
    if let Some(pos) = url.find("://") {
        if let Some(at_pos) = url[pos + 3..].find('@') {
            let prefix = &url[..pos + 3];
            let suffix = &url[pos + 3 + at_pos + 1..];
            return format!("{}***@{}", prefix, suffix);
        }
    }
    url.to_string()
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 初始化日志
    tracing_subscriber::registry()
        .with(tracing_subscriber::fmt::layer().with_target(false))
        .with(EnvFilter::from_default_env().add_directive("rustcloud_bus=info".parse()?))
        .init();

    match demonstrate_postgresql_storage().await {
        Ok(()) => {
            println!("\n💡 提示:");
            println!("   • 设置 RUST_LOG=debug 查看详细日志");
            println!("   • 设置 DATABASE_URL 连接真实数据库");
        }
        Err(e) => {
            println!("\n❌ 演示失败: {}", e);
            println!("\n💡 常见问题解决:");
            println!("   1. 确保PostgreSQL服务正在运行");
            println!("   2. 检查数据库连接参数");
            println!("   3. 确认数据库用户权限");
            println!("   4. 检查网络连接");
        }
    }

    Ok(())
}